package com.wngbms.netty.gatewaydemo.handler.bizhandler;

import com.wngbms.netty.gatewaydemo.filter.HttpRequestFilter;
import com.wngbms.netty.gatewaydemo.filter.HttpResponseFilter;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.http.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.protocol.HTTP;
import org.apache.http.util.EntityUtils;

import java.util.List;

@Slf4j
public class HttpClient4DoHttpInboundHandler extends BaseDoHttpInboundHandler {

    private CloseableHttpAsyncClient httpClient;

    public HttpClient4DoHttpInboundHandler(List<String> backends,
                                           HttpResponseFilter httpResponseFilter) {
        super(backends, httpResponseFilter);

        int cores = Runtime.getRuntime().availableProcessors();

        IOReactorConfig config = IOReactorConfig.custom()
                .setConnectTimeout(1000)
                .setSoTimeout(1000)
                .setIoThreadCount(cores)
                .setRcvBufSize(32 * 1024)
                .build();

        httpClient = HttpAsyncClients.custom()
                .setMaxConnTotal(5)
                .setMaxConnPerRoute(3)
                .setDefaultIOReactorConfig(config)
                .setKeepAliveStrategy((resp, cxt) -> 6000)
                .build();
        httpClient.start();
    }

    @Override
    public void handle(final FullHttpRequest fullHttpRequest,
                       final ChannelHandlerContext ctx,
                       HttpRequestFilter filter) {
        String backendUrl = router.route(this.backendUrls);
        filter.filter(fullHttpRequest, ctx);
        String url = backendUrl + fullHttpRequest.uri();

        proxyService.submit(() -> fetchGet(fullHttpRequest, ctx, url));
    }

    @Override
    public void fetchGet(FullHttpRequest request, ChannelHandlerContext ctx, String url) {
        HttpGet httpGet = new HttpGet(url);
        httpGet.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);
        httpGet.setHeader("system", request.headers().get("system"));

        httpClient.execute(httpGet, new FutureCallback<HttpResponse>() {
            @Override
            public void completed(HttpResponse httpResponse) {
                handleResponse(request, ctx, httpResponse);
            }

            @Override
            public void failed(Exception e) {
                httpGet.abort();
                log.error("exp", e);
            }

            @Override
            public void cancelled() {
                httpGet.abort();
            }
        });
    }

    @Override
    public void handleResponse(FullHttpRequest fullHttpRequest,
                                ChannelHandlerContext ctx,
                                Object httpResp) {
        HttpResponse httpResponse = (HttpResponse) httpResp;
        FullHttpResponse response = null;
        try {
            byte[] body = EntityUtils.toByteArray(httpResponse.getEntity());
            response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.wrappedBuffer(body));
            response.headers().set("Content-Type", "application/json");
            response.headers().setInt("Content-Length",
                    Integer.parseInt(httpResponse.getFirstHeader("Content-Length").getValue()));

            if (httpResponseFilter != null) {
                httpResponseFilter.filter(response);
            }
        } catch (Exception e) {
            log.error("exp", e);
            response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.NO_CONTENT);

            ctx.close();
        } finally {
            if (fullHttpRequest != null) {
                if (HttpUtil.isKeepAlive(fullHttpRequest)) {
                    ctx.writeAndFlush(response);
                } else {
                    ctx.writeAndFlush(response)
                            .addListener(ChannelFutureListener.CLOSE);
                }
            }
        }
    }

}
